package com.chb.flink.sink

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}

/**
 * 将wordCount的结果写到redis中
 */
object RedisSinkDemo {
    def main(args: Array[String]): Unit = {
        //初始化Flink的Streaming（流计算）上下文执行环境
        val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
        streamEnv.setParallelism(1)
        //导入隐式转换
        import org.apache.flink.streaming.api.scala._
        //读取数据
        val stream = streamEnv.socketTextStream("192.168.80.15", 8888)
        //转换计算
        val result = stream.flatMap(_.split(" "))
            .map((_, 1))
            .keyBy(0)
            .sum(1)


        // redis sink
        // 设置redis配置
        val redisConfig = new FlinkJedisPoolConfig.Builder()
            .setDatabase(1)
            .setHost("192.168.80.15")
            .setPort(6379)
            .build()

        // 创建redis Sink
        val redisSink = new RedisSink[(String, Int)](redisConfig, new RedisMapper[(String, Int)] {
            override def getCommandDescription: RedisCommandDescription = {
                new RedisCommandDescription(RedisCommand.HSET, "chb_wc")
            }

            override def getKeyFromData(t: (String, Int)): String = t._1

            override def getValueFromData(t: (String, Int)): String = t._2 + ""
        })

        // 添加redis sink
        result.addSink(redisSink)

        streamEnv.execute()
    }
}